﻿//using System;
//using System.Collections.Generic;
//using System.Linq;
//using System.Text;

//namespace ActiveMQ.Producer
//{
//    class Program
//    {
//        static void Main(string[] args)
//        {
//            var manager = new Manager();
//            manager.FetchMasterData();
//            Console.Read();
//        }

//        static void testBasic()
//        {
//            var factory = new ConnectionFactory("tcp://localhost:61616");

//            using (IConnection connection = factory.CreateConnection())
//            {
//                using (var session = connection.CreateSession())
//                {

//                    var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                    var producer = session.CreateProducer(dest);
//                    for (int i = 0; i < 40000; i++)
//                    {
//                        //var message = producer.CreateObjectMessage(new UserInfo
//                        //{
//                        //    ID = i + 1,
//                        //    Username = Guid.NewGuid().ToString()
//                        //});
//                        //message.Properties.SetString("property", "property");

//                        var textMsg = producer.CreateTextMessage(i.ToString().PadLeft(5, '0'));
//                        producer.Send(textMsg);
//                        //Thread.Sleep(500);
//                        Console.WriteLine("send " + i.ToString() + " success");
//                    }
//                }
//            }
//        }

//        static void sendUserInfo()
//        {
//            var helper = new ActiveMqProducerHelper();
//            helper.Open();
//            for (var i = 0; i < 1000; i++)
//            {
//                helper.SendObjectMessage("first_queue", new UserInfo { ID = i + 1, Username = "test" + i.ToString() });
//                Console.WriteLine("send ok");
//            }
//            helper.Close();
//        }

//        public class ActiveMqMessageWrapper<T>
//        {
//            public T Message { get; set; }
//            public Dictionary<string, string> Properties { get; set; }

//            public ActiveMqMessageWrapper()
//            {
//                Properties = new Dictionary<string, string>();
//            }
//        }

//        public class ActiveMqConsumerHelper
//        {
//            public ISession Session { get; set; }
//            public IConnection Connection { get; set; }
//            private object syncObj = new object();

//            private Dictionary<string, IMessageConsumer> _consumers;


//            public void Open()
//            {
//                Connection = ConnectionUtil.CreateSession();
//                Connection.Start();
//                Session = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);

//                _consumers = new Dictionary<string, IMessageConsumer>(StringComparer.OrdinalIgnoreCase);

//            }

//            public IMessageConsumer GetConsumer(String destinationName)
//            {

//                if (!_consumers.ContainsKey(destinationName))
//                {
//                    lock (syncObj)
//                    {
//                        IMessageConsumer consumer = null;
//                        if (!_consumers.TryGetValue(destinationName, out consumer))
//                        {
//                            var destination = Session.GetDestination(destinationName);
//                            consumer = Session.CreateConsumer(destination);
//                            _consumers.Add(destinationName, consumer);
//                        }
//                        return consumer;
//                    }
//                } else
//                {
//                    return _consumers[destinationName];
//                }

//            }

//            public void Close()
//            {
//                try
//                {
//                    Session.Close();
//                } finally { }

//                try
//                {
//                    Connection.Close();
//                } finally { }

//                foreach (var consumer in _consumers)
//                {
//                    try
//                    {
//                        consumer.Value.Close();
//                    } finally { }
//                }
//            }

//            public ActiveMqMessageWrapper<string> ReceiveTextMessage(string destinationName)
//            {
//                var consumer = GetConsumer(destinationName);
//                var message = consumer.Receive(TimeSpan.FromMilliseconds(100));
//                //var message = consumer.ReceiveNoWait();

//                if (message == null)
//                {
//                    //Console.WriteLine("destinationName [{0}] message null", destinationName);
//                    return null;
//                }
//                var textMessage = message as ITextMessage;
//                var text = textMessage.Text;
//                var wrapper = new ActiveMqMessageWrapper<string>();
//                wrapper.Message = text;
//                foreach (String key in textMessage.Properties.Keys)
//                {
//                    wrapper.Properties.Add(key, textMessage.Properties.GetString(key));
//                }
//                return wrapper;
//            }


//            public ActiveMqMessageWrapper<T> ReceiveObjectMessage<T>(string destinationName)
//            {
//                var consumer = GetConsumer(destinationName);
//                var objectMessage = consumer.Receive(TimeSpan.Zero);
//                if (objectMessage == null)
//                    return null;

//                var message = objectMessage as IObjectMessage;
//                var wrapper = new ActiveMqMessageWrapper<T>();
//                wrapper.Message = (T)message.Body;
//                foreach (String key in message.Properties.Keys)
//                {
//                    wrapper.Properties.Add(key, message.Properties.GetString(key));
//                }
//                return wrapper;
//            }

//        }

//        public class ActiveMqProducerHelper
//        {
//            public ISession Session { get; set; }
//            public IConnection Connection { get; set; }
//            private object syncObj = new object();

//            private Dictionary<string, IMessageProducer> _producers;


//            public void Open()
//            {
//                Connection = ConnectionUtil.CreateSession();
//                Session = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);

//                _producers = new Dictionary<string, IMessageProducer>(StringComparer.OrdinalIgnoreCase);

//            }

//            public IMessageProducer GetProducer(String destinationName)
//            {

//                if (!_producers.ContainsKey(destinationName))
//                {
//                    lock (syncObj)
//                    {
//                        IMessageProducer producer = null;
//                        if (!_producers.TryGetValue(destinationName, out producer))
//                        {
//                            var destination = Session.GetDestination(destinationName);
//                            producer = Session.CreateProducer(destination);
//                            _producers.Add(destinationName, producer);
//                        }
//                        return producer;
//                    }
//                } else
//                {
//                    return _producers[destinationName];
//                }

//            }

//            public void Close()
//            {
//                try
//                {
//                    Session.Close();
//                } finally { }

//                try
//                {
//                    Connection.Close();
//                } finally { }

//                foreach (var producer in _producers)
//                {
//                    try
//                    {
//                        producer.Value.Close();
//                    } finally { }
//                }
//            }


//            public void SendObjectMessage(string destinationName, Object message)
//            {
//                var producer = GetProducer(destinationName);
//                var objectMessage = Session.CreateObjectMessage(message);
//                producer.Send(objectMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.MinValue);
//            }

//            public void SendTextMessage(string destinationName, string message, Dictionary<string, string> properties)
//            {
//                var producer = GetProducer(destinationName);

//                var textMessage = Session.CreateTextMessage(message);
//                if (properties != null)
//                {
//                    foreach (var kvp in properties)
//                    {
//                        textMessage.Properties.SetString(kvp.Key, kvp.Value);
//                    }
//                }
//                producer.Send(textMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.MinValue);
//            }
//        }

//        public class ConnectionUtil
//        {
//            private readonly static ConnectionFactory ConnectionFactory;
//            private static readonly String url = "tcp://localhost:61616";
//            static ConnectionUtil()
//            {
//                ConnectionFactory = new ConnectionFactory(url);
//            }

//            public static IConnection CreateSession()
//            {
//                return ConnectionFactory.CreateConnection();
//            }

//        }

//        public interface IFetcher
//        {
//            void Fetch();
//            List<string> FetchGetDocList();

//            ActiveMqProducerHelper ActiveMqProducerHelper { get; set; }
//            ActiveMqConsumerHelper ActiveMqConsumerHelper { get; set; }
//            string MasterDataType { get; set; }
//            string GetDocListDestinationName { get; set; }
//            IGetDocListFetcher GetDocListFetcher { get; set; }

//        }

//        public interface IGetDocListFetcher
//        {
//            List<String> GetDocList(String mdType, string TCode, List<string> orgs);
//        }

//        public class DefaultGetDocListFetcher : IGetDocListFetcher
//        {
//            public List<string> GetDocList(String mdType, string TCode, List<string> orgs)
//            {
//                var list = new List<string>();
//                Console.WriteLine("begin getdoclist");
//                for (int i = 1; i <= 20; i++)
//                {
//                    list.Add(i.ToString().PadLeft(5, '0'));

//                    //Thread.Sleep(100);
//                }
//                Console.WriteLine("end getdoclist");
//                return list;
//            }
//        }

//        public class PIFetcher : IFetcher
//        {
//            //public ConcurrentQueue<String> GetDocList { get; set; }

//            //public ConcurrentQueue<UserInfo> UserInfos { get; set; }
//            //public ManualResetEvent Event { get; set; }
//            public ActiveMqProducerHelper ActiveMqProducerHelper { get; set; }
//            public ActiveMqConsumerHelper ActiveMqConsumerHelper { get; set; }
//            public string DetailDestinationName { get; set; }
//            public string GetDocListDestinationName { get; set; }
//            public string MasterDataType { get; set; }
//            public IGetDocListFetcher GetDocListFetcher { get; set; }

//            //public PIFetcher()
//            //{
//            //    //this.InitParam();
//            //}

//            //public UserInfo RequestParam { get; set; }

//            //public void InitParam()
//            //{
//            //    RequestParam = new UserInfo() { Username = "init", ID = 0 };
//            //}

//            public void Fetch()
//            {

//                //UserInfo userInfo = null;
//                int count = 0;
//                //while (UserInfos.TryDequeue(out userInfo))
//                //{
//                //    Console.WriteLine("Thread: {0}, {1}... Fetch data", Thread.CurrentThread.ManagedThreadId, userInfo);
//                //    ActiveMqProducerHelper.SendObjectMessage(DetailDestinationName, userInfo);
//                //    count++;
//                //    Thread.Sleep(350); // 模拟网络操作
//                //}
//                ActiveMqMessageWrapper<string> message = null;
//                ActiveMqConsumerHelper consumer = new ActiveMqConsumerHelper();
//                ActiveMqProducerHelper producer = new ActiveMqProducerHelper();

//                consumer.Open();
//                producer.Open();
//                do
//                {
//                    message = consumer.ReceiveTextMessage(this.GetDocListDestinationName);
//                    if (message != null)
//                    {
//                        Console.WriteLine("[Thread: {0}] receive getdoclist {1}, search detail", Thread.CurrentThread.ManagedThreadId, message.Message);
//                        var md = this.MasterDataType;
//                        producer.SendObjectMessage(DetailDestinationName, new UserInfo
//                        {
//                            ID = 0,
//                            Username = md,
//                            Code = message.Message
//                        });
//                        count++;
//                        Thread.Sleep(50);
//                        // 查询详细操作
//                    }


//                } while (message != null);
//                consumer.Close();
//                producer.Close();

//                Console.WriteLine("Thread: {0}, process total: {1}", Thread.CurrentThread.ManagedThreadId, count.ToString());

//            }

//            public void Fetch2()
//            {
//                String docNo = string.Empty;
//                int count = 0;
//                //while (GetDocList.TryDequeue(out docNo))
//                //{
//                //    Console.WriteLine("Thread: {0}, Fetch data {1}", Thread.CurrentThread.ManagedThreadId, docNo);
//                //    RequestParam.Code = docNo;
//                //    ActiveMqProducerHelper.SendObjectMessage(DetailDestinationName, RequestParam);
//                //    count++;
//                //    Thread.Sleep(350); // 模拟网络操作
//                //}
//                ActiveMqMessageWrapper<string> message = null;
//                do
//                {
//                    message = ActiveMqConsumerHelper.ReceiveTextMessage(this.GetDocListDestinationName);
//                    Console.WriteLine("[Thread: {0}] receive getdoclist {0}, search detail", Thread.CurrentThread.ManagedThreadId, message.Message);
//                    // 查询详细操作
//                    var md = this.MasterDataType;
//                    ActiveMqProducerHelper.SendObjectMessage(DetailDestinationName, new UserInfo
//                    {
//                        ID = 0,
//                        Username = md,
//                        Code = message.Message
//                    });

//                } while (message != null);
//                Console.WriteLine("Thread: {0}, process total: {1}", Thread.CurrentThread.ManagedThreadId, count.ToString());
//            }

//            public List<string> FetchGetDocList()
//            {
//                var getdoclist = GetDocListFetcher.GetDocList(this.MasterDataType, "MM03A", new List<string>());
//                return getdoclist;
//            }
//        }

//        public class Manager
//        {
//            public ConcurrentQueue<UserInfo> UserInfos { get; set; }
//            public List<IFetcher> Fetchers { get; set; }
//            public ActiveMqProducerHelper ActiveMqProducerHelper { get; set; }
//            public ActiveMqConsumerHelper ActiveMqConsumerHelper { get; set; }
//            public IGetDocListFetcher GetDocListFetcer { get; set; }

//            //public ConcurrentQueue<string> GetDocList { get; set; }

//            public void InitGetDocList()
//            {
//                GetDocListFetcer = new DefaultGetDocListFetcher();
//            }

//            public void InitFetchers()
//            {
//                Fetchers = new List<IFetcher>();
//                Fetchers.Add(new PIFetcher()
//                {
//                    DetailDestinationName = "md.masterdata",
//                    GetDocListDestinationName = "md.masterdata.getdoclist",
//                    MasterDataType = "Material"
//                });
//            }


//            public void InitActiveMqHelper()
//            {
//                ActiveMqProducerHelper = new ActiveMqProducerHelper();
//                ActiveMqConsumerHelper = new ActiveMqConsumerHelper();
//            }

//            public void FetchMasterData()
//            {
//                int threadNums = 10;
//                this.InitGetDocList();
//                this.InitFetchers();
//                this.InitActiveMqHelper();

//                ActiveMqProducerHelper.Open();
//                ActiveMqConsumerHelper.Open();


//                Stopwatch watch = new Stopwatch();
//                foreach (var fetcher in Fetchers)
//                {
//                    fetcher.GetDocListFetcher = GetDocListFetcer;
//                    //var properties = new Dictionary<string, string>();
//                    //properties.Add("mdtype", this.MasterDataType);
//                    //var getdoclist = GetDocListFetcher.GetDocList(this.MasterDataType, "MM03A", new List<string>());
//                    //foreach (var item in getdoclist)
//                    //{
//                    //    ActiveMqProducerHelper.SendTextMessage(this.GetDocListDestinationName, item,
//                    //        properties);
//                    //    Console.WriteLine("send getdoclist to mq topic: {0}, mdtype: {1},  item: {2}",
//                    //        this.GetDocListDestinationName, this.MasterDataType, item);
//                    //}

//                    var properties = new Dictionary<string, string>();
//                    var getdocList = fetcher.FetchGetDocList();
//                    foreach (var item in getdocList)
//                    {
//                        ActiveMqProducerHelper.SendTextMessage(fetcher.GetDocListDestinationName, item,
//                            properties);
//                        Console.WriteLine("send getdoclist to mq topic: {0}, mdtype: {1},  item: {2}",
//                            fetcher.GetDocListDestinationName, fetcher.MasterDataType, item);
//                    }


//                    watch.Reset();

//                    //this.InitMockGetDocListData();

//                    //fetcher.UserInfos = UserInfos;
//                    //fetcher.ActiveMqProducerHelper = ActiveMqProducerHelper;
//                    //fetcher.ActiveMqConsumerHelper = ActiveMqConsumerHelper;

//                    List<ManualResetEvent> events = new List<ManualResetEvent>();
//                    var resetEvent = new ManualResetEvent(false);
//                    events.Add(resetEvent);
//                    watch.Start();
//                    Console.WriteLine("begin mutiple thread receive data from mq");
//                    for (int i = 1; i <= threadNums; i++)
//                    {
//                        ThreadPool.QueueUserWorkItem(obj =>
//                        {
//                            var currentFetcher = obj as PIFetcher;
//                            currentFetcher.Fetch();
//                            resetEvent.Set();
//                        }, fetcher);
//                    }

//                    WaitHandle.WaitAll(events.ToArray());
//                    watch.Stop();
//                    Console.WriteLine("all thread complted elapsed {0} ms", watch.ElapsedMilliseconds.ToString());
//                }
//                ActiveMqProducerHelper.Close();
//                ActiveMqConsumerHelper.Close();
//            }

//            public void FetchMasterData2()
//            {
//                int threadNums = 10;
//                this.InitFetchers();
//                this.InitActiveMqHelper();
//                ActiveMqProducerHelper.Open();

//                List<ManualResetEvent> events = new List<ManualResetEvent>();
//                Stopwatch watch = new Stopwatch();
//                foreach (var fetcher in Fetchers)
//                {

//                    watch.Reset();

//                    //this.InitMockGetDocListData();

//                    //fetcher.UserInfos = UserInfos;
//                    //fetcher.GetDocList = GetDocListFetcer;
//                    fetcher.ActiveMqProducerHelper = ActiveMqProducerHelper;


//                    var resetEvent = new ManualResetEvent(false);
//                    events.Add(resetEvent);
//                    watch.Start();
//                    for (int i = 1; i <= threadNums; i++)
//                    {
//                        ThreadPool.QueueUserWorkItem(obj =>
//                        {
//                            var currentFetcher = obj as PIFetcher;
//                            currentFetcher.Fetch2();
//                            resetEvent.Set();
//                        }, fetcher);
//                    }

//                    WaitHandle.WaitAll(events.ToArray());
//                    watch.Stop();
//                    Console.WriteLine("all thread complted elapsed {0} ms", watch.ElapsedMilliseconds.ToString());
//                }
//                ActiveMqProducerHelper.Close();
//            }

//        }
//    }
//}


//static void mutiple_thread_send()
//        {
//            var sender = new TestSender();
//            sender.Init();
//            var threadNums = 10;
//            for (int i = 1; i <= threadNums; i++)
//            {
//                ThreadPool.QueueUserWorkItem(state =>
//                {
//                    sender.send();
//                });
//            }
//        }

//        class TestSender
//        {
//            public IConnection Connection { get; set; }
//            public ConcurrentQueue<string> LocalQueue { get; set; }

//            public void Init()
//            {
//                var factory = new ConnectionFactory("tcp://localhost:61616");
//                Connection = factory.CreateConnection();
//                LocalQueue = new ConcurrentQueue<string>();
//                for (int i = 0; i <= 10000; i++)
//                {
//                    LocalQueue.Enqueue(i.ToString().PadLeft(5, '0'));
//                }
//            }

//            public void send()
//            {
//                var session = Connection.CreateSession();
//                var dest = session.GetDestination("first_queue", DestinationType.Queue);
//                var producer = session.CreateProducer(dest);

//                string s = string.Empty;
//                while (LocalQueue.TryDequeue(out s))
//                {
//                    var textMsg = producer.CreateTextMessage(s);
//                    producer.Send(textMsg);
//                    Console.WriteLine("[thread-{0}] send {1} success",
//                        Thread.CurrentThread.ManagedThreadId, s);
//                }
//            }
//        }